<?php

class Queue {
  private static $client = [];
  protected $name = null;

  protected function client($host = 'q1', $retry = 0) {
    if (!isset(self::$client[$host]) || is_null(self::$client[$host])) {
      self::$client[$host] = (ON_PRODUCTION) ? new Redis() : new BlackHole();
      $config = Config::get('env.redis.' . $host);//Config::get('env.redis.q')
      self::$client[$host]->pconnect($config['host'], $config['port'], 0.3);
      try {
        self::$client[$host]->select(1);	//不手动指定一下经常莫名的Queue读不出来问题，尝试去掉一次没成功，已经第二次加回来了。
      } catch (Exception $e) {
        $retry++;
        usleep(1000);
        if ($retry > 3) {
          throw $e;
        }
        return $this->client($host, $retry);
      }
    }
    return self::$client[$host];
  }

  protected function redisId() {
    return "Queue::{$this->name}";
  }

  public function __construct($name) {
    $this->name = $name;
  }

  /**
   * @param Array $item 需要push进队列的元素，必须是一个可以转成字符串的变量
   * @return mixed 如果成功，返回新队列的长度，否则返回 false
   */
  public function push(Array $item) {
    $item['action'] = $this->name;

    if (!ON_PRODUCTION) {
      include_once(ROOT . '/cronjob/consumer/' . $this->name . '.php');
      (new $this->name())->consume($item);
      //(new $this->name())->doWorker();
      return true;
    }

    //return $this->client()->rPush($this->redisId(), serialize($item));

    $res = false;
    $i = 0;
    $host = 'q1';
    while (!$res) {
      $msg = '';
      try {
        $this->client($host)->setTimeout($this->redisId(), 86400 * 10);
        $res = $this->client($host)->rPush($this->redisId(), serialize($item));
        if ($res) {
          break;
        }
      } catch (Exception $e) {
        $msg = $e->getMessage();
      }

      $i++;

      //if ($i > 1) $host = 'q1';
      if ($i > 2) usleep(30000);
      if ($i > 3) {
        Log::set('QueuePushFail', $msg . print_r($res,true) . print_r($item,true));
        return NULL;
      }
    }
    return $res;
  }

  /**
   * @return mixed 如果执行成功返回STRING，否则返回 FALSE。
   */
  public function pop() {
    try {
      if (($msg = $this->client('q1')->lPop($this->redisId()))) {
        return unserialize($msg);
      } else {
        return FALSE;
      }
    } catch (Exception $e) {
      return FALSE;
    }
  }

  /**
   * @return bool 队列是否为空。true表示队列为空。
   */
  public function isEmpty() {
    return $this->size() === 0;
  }

  /**
   * @return int 队列的长度
   */
  public function size($retry = 0){
    $s1 = '';
    try {
      $s1 = $this->client('q1')->lLen($this->redisId());
    } catch (Exception $e) {
      $retry++;
      usleep(1000);
      if ($retry > 3) {
        Log::set('Queue',$e->getMessage());
      }
      $this->size($retry);
    }
    return $s1;
  }
}
